package com.hyt.it.ogt.ksmonitor.statistics.queue;

import com.hyt.it.ogt.ksmonitor.statistics.executor.StatisticsExecutorFactory;
import com.hyt.it.ogt.ksmonitor.statistics.executor.StatisticsExecutorThreadFactory;
import com.hyt.it.ogt.ksmonitor.statistics.mq.StatisticsProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author 朱振宇
 * @since 2022-12-02
 */
@Slf4j
@Component
public class StatisticsBlockQueue implements InitializingBean {

    @Resource
    private StatisticsProducer statisticsProducer;

    protected final static BlockingQueue statisticsQueue = new LinkedBlockingQueue<>(8192);

    private static final long WAIT_TIME = 50;

    private static final int BATCH_SIZE = 100;

    @Override
    public void afterPropertiesSet() throws Exception {
        new StatisticsExecutorThreadFactory("StatisticsThreadFactory").newThread(new StatisticsTask()).start();
    }

    public static void producer(Object obj) {
        try {
            statisticsQueue.put(obj);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    final class StatisticsTask implements Runnable {
        @Override
        public void run() {
            List objList = null;
            Object obj;
            while (true) {
                try {
                    obj = statisticsQueue.take();
                    objList = new ArrayList<>(BATCH_SIZE);
                    objList.add(obj);
                    while (objList.size()  < BATCH_SIZE && Objects.nonNull((obj = statisticsQueue.poll(WAIT_TIME, TimeUnit.MILLISECONDS)))) {
                        objList.add(obj);
                    }
                } catch (InterruptedException e) {
                    log.error("考试数据统计异常");
                    e.printStackTrace();
                }
                if (Objects.nonNull(objList) && !objList.isEmpty()) {
                    try {
                        objList.forEach(vo -> StatisticsExecutorFactory.getThreadPoolExecutor().execute(() -> statisticsProducer.sendMessage(vo)));
                    } catch (Exception e) {
                        log.error("考试数据统计异常");
                        e.printStackTrace();
                    }
                }
            }
        }
    }

}
